package com.pisolution.scheduler.remote;

import com.pisolution.scheduler.remote.codec.CommandDecoder;
import com.pisolution.scheduler.remote.codec.CommandEncoder;
import com.pisolution.scheduler.remote.command.Command;
import com.pisolution.scheduler.remote.command.CommandType;
import com.pisolution.scheduler.remote.handler.NettyClientHandler;
import com.pisolution.scheduler.remote.processor.NettyRequestProcessor;
import com.pisolution.scheduler.remote.tracker.RequestTracker;
import com.pisolution.scheduler.remote.utils.Host;
import com.pisolution.scheduler.remote.utils.NettyUtils;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nullable;
import java.net.InetSocketAddress;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 节点组件之间信息交互的客户端.
 *
 * @author sora
 * @version 0.1.0
 */
// TODO: 确认所有方法是否已被调用.
public final class NettyRemoteClient {
    private static final Logger LOGGER = LoggerFactory.getLogger(NettyRemoteClient.class);

    // -------------------------------------------------------------------------------------------------

    /** 用于配置并启动客户端. */
    private final Bootstrap bootstrap;

    /** 用于处理所有事件的事件组. */
    private final EventLoopGroup group;

    /** 用于定期移除超时的请求. */
    private final ScheduledExecutorService executor;

    /** 用于路由响应到适配的处理器. */
    private final NettyClientHandler handler;

    /** 用于编码交互命令. */
    private final CommandEncoder encoder;

    /** 用于存储已经创建的服务端连接. */
    private final ConcurrentHashMap<Host, Channel> channels;

    // -------------------------------------------------------------------------------------------------

    /** 用于确认客户端程序是否已经启动. */
    private final AtomicBoolean isStarted = new AtomicBoolean(false);

    // -------------------------------------------------------------------------------------------------
    //  初始化
    // -------------------------------------------------------------------------------------------------

    public NettyRemoteClient() {
        this(Runtime.getRuntime().availableProcessors());
    }

    /**
     * 初始化客户端配置.
     *
     * @param threads 事件组可用线程数量.
     */
    public NettyRemoteClient(final int threads) {
        // 1. 初始化客户端配置.
        this.bootstrap = new Bootstrap();
        this.executor = Executors.newSingleThreadScheduledExecutor();
        this.handler = new NettyClientHandler(this);
        this.encoder = new CommandEncoder();
        this.channels = new ConcurrentHashMap<>(128);

        // 2. 如果epoll技术可用 则启用该技术.
        if (NettyUtils.useEpoll()) {
            this.group = new EpollEventLoopGroup(threads);

            return;
        }

        this.group = new NioEventLoopGroup(threads);

        // 3. 启动客户端.
        this.start();
    }

    // -------------------------------------------------------------------------------------------------
    //  生命周期
    // -------------------------------------------------------------------------------------------------

    /**
     * 配置并启动客户端.
     */
    private void start() {
        // 1. 配置客户端.
        this.bootstrap
            .group(this.group)
            .channel(NettyUtils.useEpoll() ? EpollSocketChannel.class : NioSocketChannel.class)
            .option(ChannelOption.SO_KEEPALIVE, true)
            .option(ChannelOption.TCP_NODELAY, true)
            .option(ChannelOption.SO_SNDBUF, 65535)
            .option(ChannelOption.SO_RCVBUF, 65535)
            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000)
            .handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) {
                    ch.pipeline().addLast(new CommandDecoder(), handler, encoder);
                }
            });

        // 2. 固定频率移除过期请求.
        this.executor.scheduleAtFixedRate(RequestTracker::removeTimeout, 5000, 1000, TimeUnit.MILLISECONDS);

        // 3. 更新启动标识.
        this.isStarted.compareAndSet(false, true);
    }

    /**
     * 关闭客户端.
     */
    public void stop() {
        if (this.isStarted.compareAndSet(true, false)) {
            try {
                // 1. 关闭所有服务端连接.
                for (final Channel channel: this.channels.values()) {
                    channel.close();
                }
                this.channels.clear();

                // 2. 关闭事件组.
                if (this.group != null) {
                    this.group.shutdownGracefully();
                }

                // 3. 关闭过期请求清除定时器.
                if (this.executor != null) {
                    this.executor.shutdownNow();
                }
            } catch (Exception e) {
                LOGGER.error("Could not close the netty client.", e);
            }

            LOGGER.info("The netty client closed.");
        }
    }

    // -------------------------------------------------------------------------------------------------
    //  发送请求
    // -------------------------------------------------------------------------------------------------

    /**
     * 向指定主机上的服务端发送请求.
     *
     * @param host 启动了服务端的主机地址.
     * @param command 待发送到该服务端的请求命令.
     */
    public void send(final Host host, final Command command) {
        // 1. 获取缓存中与该主机之间建立的连接.
        final Channel channel = this.getChannel(host);

        // 2. 校验连接是否可用.
        if (channel == null) {
            throw new RuntimeException(String.format("Could not connect to the host %s.", host.getAddress()));
        }

        try {
            // 3. 发送命令到服务端.
            final ChannelFuture future = channel.writeAndFlush(command).await();

            // 4. 确认请求是否成功.
            if (future.isSuccess()) {
                LOGGER.debug("Succeed to send the command {} to the host {}.", command, host.getAddress());

                return;
            }

            final String errorMsg = String.format("Could not send the command %s to the host %s.", command, host.getAddress());

            LOGGER.error(errorMsg, future.cause());
            throw new RuntimeException(errorMsg);
        } catch (Exception e) {
            throw new RuntimeException(
                String.format("Could not send the command %s to the host %s.", command, host.getAddress()),
                e
            );
        }
    }

    /**
     * 以同步的方式向指定主机上的服务端发送请求.
     *
     * @param host 启动了服务端的主机地址.
     * @param command 待发送到该服务端的请求命令.
     * @param timeout 处理该命令所需的时间限制.
     *
     * @return 服务端对于该请求命令的响应.
     *
     * @throws InterruptedException 如果该请求被其他请求打断.
     */
    public Command send(final Host host, final Command command, final long timeout) throws InterruptedException {
        // 1. 获取缓存中与该主机之间建立的连接.
        final Channel channel = this.getChannel(host);

        // 2. 校验连接是否可用.
        if (channel == null) {
            throw new RuntimeException(String.format("Could not connect to the host %s.", host.getAddress()));
        }

        // 3. 发送请求到服务端.
        final RequestTracker tracker = new RequestTracker(command.getOpaque(), timeout);

        channel.writeAndFlush(command).addListener((channelFuture) -> {
            if (channelFuture.isSuccess()) {
                tracker.setSentOk(true);

                return;
            }

            tracker.setSentOk(false);
            tracker.setCause(channelFuture.cause());

            LOGGER.error("Could not send the command {} to the host {}.", command, host.getAddress());
        });

        // 4. 等待请求处理完成.
        final Command response = tracker.waitResponse();

        // 5. 校验响应是否非法.
        if (response == null) {
            if (tracker.isSentOk()) {
                throw new RuntimeException(String.format(
                    "The request of the command %s towards the host %s timed out.",
                    command,
                    host.getAddress()
                ));
            }

            throw new RuntimeException(
                String.format("Could not send the command %s to the host %s.", command, host.getAddress()),
                tracker.getCause()
            );
        }

        return response;
    }

    // -------------------------------------------------------------------------------------------------
    //  处理器注册
    // -------------------------------------------------------------------------------------------------

    /**
     * 注册指定的处理器到客户端以处理相应的响应命令 处理时使用默认的执行器创建执行环境.
     *
     * @param type 该处理器所处理的响应命令类型.
     * @param processor 待注册处理器.
     *
     * @return {@link NettyRemoteClient} 对象以便于链式调用注册多个处理器.
     */
    public NettyRemoteClient register(final CommandType type, final NettyRequestProcessor processor) {
        return this.register(type, processor, null);
    }

    /**
     * 注册指定的处理器到客户端以处理相应的响应命令 处理时使用给定的执行器创建执行环境.
     *
     * @param type 该处理器所处理的响应命令类型.
     * @param processor 待注册处理器.
     * @param executor 用于创建处理器的执行环境.
     *
     * @return {@link NettyRemoteClient} 对象以便于链式调用注册多个处理器.
     */
    public NettyRemoteClient register(final CommandType type, final NettyRequestProcessor processor, final ExecutorService executor) {
        this.handler.register(type, processor, executor);

        return this;
    }

    // -------------------------------------------------------------------------------------------------
    //  连接维护
    // -------------------------------------------------------------------------------------------------

    /**
     * 关闭与给定主机之间的连接.
     *
     * @param host 需要关闭的主机连接的信息.
     */
    public void closeChannel(Host host) {
        final Channel channel = this.channels.get(host);

        if (channel != null) {
            channel.close();
        }
    }

    /**
     * 尝试获取与给定主机之间建立的连接 如果该连接不存在 则创建一个新的连接对象.
     *
     * @param host 用于确认需要索取的连接.
     *
     * @return 相应的连接对象 如果不存在 则创建一个新的连接对象.
     */
    @Nullable
    public Channel getChannel(final Host host) {
        // 1. 尝试寻找该连接.
        final Channel channel = this.channels.get(host);

        // 2. 如果该连接还存活 直接返回.
        if (channel != null && channel.isActive()) {
            return channel;
        }

        // 3. 创建一个新的连接对象.
        ChannelFuture future;
        try {
            synchronized (this.bootstrap) {
                future = this.bootstrap.connect(new InetSocketAddress(host.getIp(), host.getPort()));
            }

            // 4. 等待连接建立.
            future.sync();

            // 5. 缓存创建完成的连接.
            if (future.isSuccess()) {
                final Channel ch = future.channel();

                this.channels.put(host, ch);
                return ch;
            }
        } catch (Exception e) {
            LOGGER.warn("Could not connect to the remote server {}.", host.getAddress(), e);
        }

        return null;
    }
}
